package com.giw.project.advanced_graph_traversal;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.giw.project.exception_manager.UncheckedException;

public class TraversalReducer extends Reducer<Text, Text, Text, Text> {

	private final static String tmp_path = "tmp2.txt";
	private final static String tmp_path1 = "tmp.txt";
	private final static String tmp_pathS = "tmp_searchKind.txt";
	private final static String tmp_endSearch = "tmp_endSearch.txt";

	private final static Map<String, String> link2successors_director = new HashMap<String, String>();
	private final static Map<String, String> link2successors_genre = new HashMap<String, String>();
	private final static Map<String, String> link2kind = new HashMap<String, String>();
	private final static Map<String, String> link2successors_actors = new HashMap<String, String>();
	private final static Map<String, Map<String, String>> kind2map = new HashMap<String, Map<String, String>>();

	private int director_rank;
	private int genre_rank;
	private int actor_rank;

	protected void setup(Context context) {
		link2successors_director.put("title-<http://purl.org/dc/terms/title>",
				"<http://data.linkedmdb.org/resource" + "/movie/director>");
		link2successors_director.put(
				"movie-<http://data.linkedmdb.org/resource"
						+ "/movie/director>",
				"<http://data.linkedmdb.org/resource/movie/director>");
		link2successors_director.put(
				"director-<http://data.linkedmdb.org/resource"
						+ "/movie/director>",
				"<http://xmlns.com/foaf/0.1/page>");

		link2successors_genre.put("title-<http://purl.org/dc/terms/title>",
				"<http://data.linkedmdb.org/resource/movie/genre>");
		link2successors_genre.put("movie-<http://data.linkedmdb.org/resource"
				+ "/movie/genre>",
				"<http://data.linkedmdb.org/resource/movie/genre>");
		link2successors_genre.put("genre-<http://data.linkedmdb.org/resource"
				+ "/movie/genre>", "<http://xmlns.com/foaf/0.1/page>");

		link2successors_actors.put("title-<http://purl.org/dc/terms/title>",
				"<http://data.linkedmdb.org/resource/movie/actor>");
		link2successors_actors.put("movie-<http://data.linkedmdb.org/resource"
				+ "/movie/actor>",
				"<http://data.linkedmdb.org/resource/movie/actor>");
		link2successors_actors.put("actor-<http://data.linkedmdb.org/resource"
				+ "/movie/actor>", "<http://xmlns.com/foaf/0.1/page>");

		link2kind.put("<http://data.linkedmdb.org/resource/film/", "movie");
		link2kind.put("<http://data.linkedmdb.org/resource/director/",
				"director");
		link2kind.put("<http://data.linkedmdb.org/resource/film_genre/",
				"genre");
		link2kind.put("string", "title");
		link2kind.put("<http://data.linkedmdb.org/resource/actor/", "actor");

		kind2map.put("genre", link2successors_genre);
		kind2map.put("director", link2successors_director);
		kind2map.put("actor", link2successors_actors);
	}

	private Map<String, String> getSuccessorsMap(String genre)
			throws IOException {

		return kind2map.get(genre);

	}

	protected void reduce(Text key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {

		Map<String, Map<String, String>> currentNodes2links = new HashMap<String, Map<String, String>>();
		File f = new File(tmp_path1);
		FileInputStream fis = new FileInputStream(f);
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line = buffer.readLine();
		while (line != null) {
			String currentNode = line.split("\t")[0];
			String currentLink = line.split("\t")[1];
			String currentFlow = line.split("\t")[2];
			Map<String, String> flow2link;
			flow2link = currentNodes2links.get(currentNode);
			if (flow2link == null)
				flow2link = new HashMap<String, String>();
			flow2link.put(currentFlow, currentLink);
			currentNodes2links.put(currentNode, flow2link);
			line = buffer.readLine();
		}
		this.initializeRank();
		boolean endSearch = this.verifyEndSearch(currentNodes2links);
		buffer.close();
		is.close();
		fis.close();
		String node = key.toString();
		Map<String, String> links = currentNodes2links.get(node);
		if (links == null)
			links = currentNodes2links
					.get(node.substring(1, node.length() - 1));
		int genre_count = 0;
		int actor_count = 0;
		for (Text neighbour : values) {
			String[] neighbours = neighbour.toString().split("},");
			for (String neigh : neighbours) {
				String current_link = neigh.substring(1).split(",")[0];
				String actualNode = node;
				if (currentNodes2links.get(node) == null)
					actualNode = node.substring(1, (node.length() - 1));
				for (String genre : currentNodes2links.get(actualNode).keySet()) {

					if (currentNodes2links.get(actualNode).get(genre).equals(
							current_link)) {
						try {
							String new_node = neigh.split(",")[1];
							if (new_node.charAt(new_node.length() - 1) == '.') {
								new_node = new_node.substring(0, new_node
										.length() - 2);
							}
							if (!endSearch) {

								String kind = this.getNodeKind(node);
								String a = kind + "-" + current_link;

								if (kind.equals("movie")
										&& current_link
												.equals("<http://data.linkedmdb.org/resource/movie/genre>")
										&& genre_count < 3) {
									genre_count++;
									scriviSuFile(new_node, this
											.getSuccessorsMap(genre).get(a),
											genre);
								}
								if (kind.equals("movie")
										&& current_link
												.equals("<http://data.linkedmdb.org/resource/movie/actor>")
										&& actor_count < 5) {
									actor_count++;
									scriviSuFile(new_node, this
											.getSuccessorsMap(genre).get(a),
											genre);
								}
								else{
									scriviSuFile(new_node, this
											.getSuccessorsMap(genre).get(a),
											genre);}

							} else {
								context.write(new Text(node + "\t"
										+ this.getMovie2Rank(node).get(node)),
										new Text(new_node));
								File fS = new File(tmp_endSearch);
								fS.delete();
							}
						} catch (Exception e) {
							throw new UncheckedException(
									"Error while trying to write the temporary file!\n"
											+ "Message:" + e.getMessage());
						}
					}
				}
			}
		}
	}

	private void scriviSuFile(String node, String link, String flow)
			throws Exception {
		File f = new File(tmp_path);
		FileOutputStream fos = null;
		if (f.exists()) {
			fos = new FileOutputStream(f, true);
		} else
			fos = new FileOutputStream(f);
		PrintStream output = new PrintStream(fos);
		output.println(node + "\t" + link + "\t" + flow);
		output.close();
		fos.close();

	}

	private boolean verifyEndSearch(Map<String, Map<String, String>> nodes2links) {
		for (String node : nodes2links.keySet()) {
			if (node.length() < 41)
				return false;
			if (node.substring(0, 41).equals(
					"<http://data.linkedmdb.org/resource/film/")
					&& nodes2links.get(node).values().toArray()[0]
							.equals("<http://xmlns.com/foaf/0.1/page>")) {
				return true;
			}
			return false;
		}
		return false;
	}

	private String getNodeKind(String uri) {
		if (!uri.contains("http")) {
			return link2kind.get("string");
		}
		String[] parts_uri = uri.split("/");
		int i;
		uri = "";
		for (i = 0; i < parts_uri.length - 1; i++) {
			uri += "/" + parts_uri[i];
		}
		return link2kind.get(uri.substring(1) + "/");

	}

	private Map<String, Integer> getMovie2Rank(String uri_link)
			throws IOException {
		File f = new File(tmp_path1);
		FileInputStream fis = new FileInputStream(f);
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line = buffer.readLine();
		Map<String, Integer> movie2rank = new HashMap<String, Integer>();
		while (line != null) {
			String currentNode = line.split("\t")[0];
			if (currentNode.equals(uri_link)) {
				String currentFlow = line.split("\t")[2];
				Integer current_rank = 0;
				if (currentFlow.equals("director"))
					current_rank = this.director_rank;
				if (currentFlow.equals("genre"))
					current_rank = this.genre_rank;
				if (currentFlow.equals("actor"))
					current_rank = this.actor_rank;
				Integer count = movie2rank.get(currentNode);
				if (count == null)
					count = 0;
				count += current_rank;
				movie2rank.put(currentNode, count);
			}
			line = buffer.readLine();
		}
		buffer.close();
		is.close();
		fis.close();
		return movie2rank;

	}

	private void initializeRank() throws IOException {
		File f = new File(tmp_pathS);
		FileInputStream fis = new FileInputStream(f);
		InputStreamReader is = new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		String line = buffer.readLine();
		this.director_rank = Integer.parseInt(line);
		line = buffer.readLine();
		this.genre_rank = Integer.parseInt(line);
		line = buffer.readLine();
		this.actor_rank = Integer.parseInt(line);
		buffer.close();
		is.close();
		fis.close();
	}

}
